﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Linq.Expressions;
using System.Net.Sockets;
using System.Threading;
using Apache.Cassandra;
using Cassandraemon;
using Cassandraemon.Connection;
using Microsoft.VisualStudio.TestTools.UnitTesting;

namespace Cassandraemon.Test
{
	/// <summary>
	/// This test requires Cassandra server running on localhost,
	/// with default keyspace/column family definitions.
	/// See Server Configurations region for changing these settings.
	/// </summary>
	[TestClass]
	public partial class CassandraContextTest
	{
		#region TestContext

		private TestContext testContextInstance;

		public TestContext TestContext
		{
			get
			{
				return testContextInstance;
			}
			set
			{
				testContextInstance = value;
			}
		}

		#endregion

        #region Server Configurations

		public static CassandraContext CreateContext()
        {
            return new CassandraContext(_Host, _Port, _KeySpace);
        }

        public static void DeleteKeyAndSubmitChanges(byte[] key)
        {
            using (var ctx = CreateContext())
            {
                ctx.Column.DeleteOnSubmit(x =>
                     x.Key == key &&
                     x.ColumnFamily == _ColumnFamily);
                ctx.SubmitChanges();
            }
        }

        #endregion

        #region Keyspace/Schema

        internal const string _KeySpace = "CassandraemonTest";
        internal const string _ColumnFamily = "Standard1";
        internal const string _SuperColumnFamily = "Super1";
        internal const string _IndexColumnFamily = "Index1";
        internal const string _LongIndexColumnName = "longKey";

        [ClassInitialize]
        public static void InitializeCassandraContextTest(TestContext ctx)
        {
            CassandraContext.TraceSource.Switch.Level = SourceLevels.All;
            var listener = new ConsoleTraceListener();
            CassandraContext.TraceSource.Listeners.Add(listener);

            InitializeKeyspace();
        }

        [ClassCleanup]
        public static void Cleanup()
        {
            using (var ctx = new CassandraContext(_Host, _Port, "system"))
                DropKeyspace(ctx);
        }

        static void InitializeKeyspace()
        {
            using (var ctx = new CassandraContext(_Host, _Port, "system"))
            {
                DropKeyspace(ctx);

                var ksdef = new KsDef
                {
                    Name = _KeySpace,
                    Replication_factor = 1,
                    Strategy_class = "org.apache.cassandra.locator.SimpleStrategy",
                    Cf_defs = new List<CfDef>()
                        {
                            new CfDef {
                                Keyspace = _KeySpace,
                                Name = _ColumnFamily,
                            },
                            new CfDef {
                                Keyspace = _KeySpace,
                                Name = _SuperColumnFamily,
								Column_type = "Super",
                            },
                            new CfDef {
                                Keyspace = _KeySpace,
                                Name = _IndexColumnFamily,
                                Column_metadata = new List<ColumnDef> {
                                    new ColumnDef {
                                        Name = _LongIndexColumnName.ToCassandraByte(),
                                        Validation_class = "LongType",
                                        Index_type = IndexType.KEYS,
                                    },
                                },
                            },
                        }
                };
                var schemaVersion = ctx.SystemAddKeyspace(ksdef);
            }

            // When the ring has multiple nodes,
            // adding keyspaces may need a little time
            // to be propagated.
            Thread.Sleep(500);
        }

        static void DropKeyspace(CassandraContext ctx)
        {
            var rgksd = ctx.DescribeKeySpaces();
            if (rgksd.Any(ksd => ksd.Name == _KeySpace))
            {
                ctx.SystemDropKeyspace(_KeySpace);
            }
        }

        #endregion

        #region Utilities

        static IEnumerable<long> RangeLong(int start, int count)
        {
            return Enumerable.Range(start, count).Select(i => (long)i);
        }

        #endregion

        #region Constructors & ConnectionPool

        /// <summary>
        /// Just create CassandraContext and then dispose.
        /// </summary>
        [TestMethod]
		public void ConstructorTest()
		{
			using (var ctx = CreateContext())
			{
			}
		}

		const string HostToFail = "aHostThatNeverExist";

		/// <summary>
		/// Confirm that "HostToFail" fails to make sure ConstructorFailOverTest is valid.
		/// </summary>
		[ExpectedException(typeof(SocketException))]
		[TestMethod]
		public void ConstructorFailTest()
		{
			using (var pool = new ConnectionPool(
                HostToFail,
                _Port, _KeySpace))
			{
				using (var ctx = new CassandraContext(pool))
				{
				}
			}
		}

		[TestMethod]
		public void ConstructorFailOverTest()
		{
			for (int i = 0; i < 2; i++)
			{
				using (var pool = new ConnectionPool(
                    string.Concat(HostToFail, " ", _Host),
                    _Port, _KeySpace))
				{
					using (var ctx = new CassandraContext(pool))
					{
					}
				}
			}

			for (int i = 0; i < 2; i++)
			{
				using (var pool = new ConnectionPool(
                    string.Concat(_Host, " ", HostToFail),
                    _Port, _KeySpace))
				{
					using (var ctx = new CassandraContext(pool))
					{
					}
				}
			}
		}

		/// <summary>
		/// This test succeeds only when there are 2 or more nodes.
		/// You can ignore the failure otherwise.
		/// </summary>
		[TestMethod]
		public void LoadBalanceTest()
		{
			string host1;
			using (var ctx1 = CreateContext())
			{
                var pool = ctx1.Pool;
                if (pool.Nodes.Count == 1)
                {
                    Assert.Inconclusive("This test requires 2 or more nodes in the cluster");
                    return;
                }
				host1 = ctx1.ClientHostName;
			}
			using (var ctx2 = CreateContext())
			{
				var host2 = ctx2.ClientHostName;
				Assert.AreNotEqual(host1, host2);
			}

			using (var ctx1 = CreateContext())
			{
				host1 = ctx1.ClientHostName;
				using (var ctx2 = CreateContext())
				{
					var host2 = ctx2.ClientHostName;
					Assert.AreNotEqual(host1, host2);
				}
			}
        }

        #endregion

        #region LINQ Queries

        [TestMethod]
        public void EmptyResultTest()
        {
            using (var ctx = CreateContext())
            {
                var result = ctx.Column
                    .Where(e => e.ColumnFamily == _ColumnFamily && e.Key == long.MaxValue && e.Column == "test")
                    .FirstOrDefault();
                Assert.IsNull(result);
            }
        }

        [TestMethod]
        public void KeyEqualMethodTest()
        {
            using (var ctx = CreateContext())
            {
                var result = ctx.ColumnList
                    .Where(e => e.ColumnFamily == _ColumnFamily && e.Key == GetKey())
                    .FirstOrDefault(e => e.Data != null);
            }
        }

        [TestMethod]
        public void KeyInMethodTest()
        {
            using (var ctx = CreateContext())
            {
                var result = ctx.ColumnList
                    .Where(e => e.ColumnFamily == _ColumnFamily && e.Key.In(GetKeys()))
                    .FirstOrDefault(e => e.Data != null);
            }
        }

        static long GetKey() { return long.MaxValue; }
        static long[] GetKeys() { return new[] { long.MaxValue }; }

        #endregion

        #region Index

        [TestMethod]
        [ExpectedException(typeof(InvalidCastException))] // Index[].In() is not supported
        public void IndexInTest()
        {
            using (var ctx = CreateContext())
            {
                var keys = RangeLong(1, 100).ToArray();
                foreach (var key in keys)
                   ctx.InsertOnSubmit(_IndexColumnFamily, key, new Column().SetNameValue(_LongIndexColumnName, key));
                ctx.SubmitChanges();

                var actual = ctx.Column
                    .Where(e => e.ColumnFamily == _IndexColumnFamily && e.Index[_LongIndexColumnName].In(keys))
                    .AsEnumerable()
                    .Select(e => e.Key.ToInt64())
                    .OrderBy(i => i)
                    .ToList();
                CollectionAssert.AreEqual(keys, actual);
            }
        }

        #endregion

        #region Add/Delete (Simple)

        [TestMethod]
        public void InsertOnSubmitColumnTest()
        {
            byte[] key = "Key1".ToCassandraByte();
            const string columnName = "name1";
            const string value = "value1";

            using (var ctx = CreateContext())
            {
                ctx.Column.DeleteOnSubmit(x =>
                    x.ColumnFamily == _ColumnFamily &&
                    x.Key == key);
                ctx.SubmitChanges();

                var column = new Column()
                    .SetNameValue(columnName, value);
                ctx.InsertOnSubmit(_ColumnFamily, key, column);

                var rgcol = Enumerable.Range(0, 100)
                    .Select(i => new Column().SetNameValue(i, i));
                ctx.InsertOnSubmit(_ColumnFamily, key, rgcol);

                ctx.SubmitChanges();

                rgcol = ctx.ColumnList
                    .Where(x =>
                        x.ColumnFamily == _ColumnFamily &&
                        x.Key == key)
                    .TakeColumn(int.MaxValue)
                    .SelectMany(x => x.Data);
                Assert.AreEqual(101, rgcol.Count());

                ctx.Column.DeleteOnSubmit(x =>
                    x.ColumnFamily == _ColumnFamily &&
                    x.Key == key);
                ctx.SubmitChanges();
            }
        }

		/// <summary>
		/// Tests insert of an empty value column succeeds.
		/// </summary>
		[TestMethod]
		public void InsertEmptyValueTest()
		{
			var key = "Key1".ToCassandraByte();
			const string columnName = "name1";

			using (var ctx = CreateContext())
			{
				ctx.Column.DeleteOnSubmit(x =>
					x.ColumnFamily == _ColumnFamily &&
					x.Key == key);
				ctx.SubmitChanges();

				var column = new Column()
					.SetNameValue(columnName, new byte[0]);
				ctx.InsertOnSubmit(_ColumnFamily, key, column);

				ctx.SubmitChanges();

				var rgcol = ctx.ColumnList
					.Where(x =>
						x.ColumnFamily == _ColumnFamily &&
						x.Key == key)
					.SelectMany(x => x.Data);
				Assert.AreEqual(1, rgcol.Count());
				Assert.AreEqual(0, rgcol.First().Value.Length);

				ctx.Column.DeleteOnSubmit(x =>
					x.ColumnFamily == _ColumnFamily &&
					x.Key == key);
				ctx.SubmitChanges();
			}
		}

		/// <summary>
		/// This test can cause server to log:
		/// ERROR Thrift error occurred during processing of message.
		/// org.apache.thrift.protocol.TProtocolException: Missing version in
		/// readMessageBegin, old client?
		/// </summary>
		[TestMethod]
		[ExpectedException(typeof(Thrift.TApplicationException))]
		public void InsertNullValueTest()
		{
			var key = "Key1".ToCassandraByte();
			const string columnName = "name1";

			using (var ctx = CreateContext())
			{
				var column = new Column()
					.SetNameValue(columnName, null);
				ctx.InsertOnSubmit(_ColumnFamily, key, column);

				ctx.SubmitChanges();
			}
		}

        [TestMethod]
		public void AddDeleteColumnTest()
		{
			// Insert a data

			const string key = "Key1";
			const string columnName = "name1";
			const string value = "value1";

			// Delete the row to test

			using (var ctx = CreateContext())
			{
				ctx.Column.DeleteOnSubmit(x =>
					x.ColumnFamily == _ColumnFamily &&
					x.Key == key);
				ctx.SubmitChanges();
			}

			// Insert one column

			using (var ctx = CreateContext())
			{
				var column = new Column().SetNameValue(columnName, value);
				var entity = new CassandraEntity<Column>()
					.SetColumnFamily(_ColumnFamily)
					.SetKey(key)
					.SetData(column);
				ctx.Column.InsertOnSubmit(entity);
				ctx.SubmitChanges();
			}

			// Verify that the data was written (Column)

			Expression<Func<CassandraEntity<Column>, bool>> expcol = (x) =>
				x.ColumnFamily == _ColumnFamily &&
				x.Key == key &&
				x.Column == columnName.ToCassandraByte();
			Expression<Func<CassandraEntity<List<Column>>, bool>> exprgcol = (x) =>
				x.ColumnFamily == _ColumnFamily &&
				x.Key == key &&
				x.Column == columnName.ToCassandraByte();
			using (var ctx = CreateContext())
			{
				var rgentcol = ctx.Column
					.Where(expcol);
				Assert.AreEqual(1, rgentcol.Count());
				foreach (var entcol in rgentcol)
				{
					Assert.AreEqual(key, entcol.Key.ToUTF8());
					Assert.AreEqual(columnName, entcol.Data.Name.ToUTF8());
					Assert.AreEqual(value, entcol.Data.Value.ToUTF8());
				}
			}

			// Verify that the data was written (ColumnList)

			using (var ctx = CreateContext())
			{
				var rgentrgcol = ctx.ColumnList
					.Where(exprgcol);
				Assert.AreEqual(1, rgentrgcol.Count());
				foreach (var entrgcol in rgentrgcol)
				{
					Assert.AreEqual(key, entrgcol.Key.ToUTF8());
					Assert.AreEqual(1, entrgcol.Data.Count);
					Assert.AreEqual(columnName, entrgcol.Data[0].Name.ToUTF8());
					Assert.AreEqual(value, entrgcol.Data[0].Value.ToUTF8());
				}
			}

			// Delete the data

			using (var ctx = CreateContext())
			{
				ctx.Column.DeleteOnSubmit(expcol);
				ctx.SubmitChanges();
			}

			// Verify that the data was deleted (Column)

			using (var ctx = CreateContext())
			{
				var rgentcol = ctx.Column
					.Where(expcol);
				Assert.AreEqual(0, rgentcol.Count());

				// Verify that the data was deleted (ColumnList)

				var rgentrgcol = ctx.ColumnList
					.Where(exprgcol);
				Assert.AreEqual(1, rgentrgcol.Count());
				Assert.AreEqual(0, rgentrgcol.First().Data.Count);
			}
		}

		[TestMethod]
		public void AddDeleteRowsTest()
		{
			const int cKeys = 10;
			const string keyPrefix = "Key";
			const string columnName = "name1";
			const string valuePrefix = "value";
			var keyMin = string.Concat(keyPrefix, "0");
			var keyMax = string.Concat(keyPrefix, cKeys - 1);

			// Insert data

			var sw = Stopwatch.StartNew();
			var column = new Column().SetName(columnName);
			for (int iKey = 0; iKey < cKeys; iKey++)
			{
				using (var ctx = CreateContext())
				{
					var key = string.Concat(keyPrefix, iKey);
					var value = string.Concat(valuePrefix, iKey);
					column.SetValue(value);
					var entity = new CassandraEntity<Column>()
						.SetColumnFamily(_ColumnFamily)
						.SetKey(key)
						.SetData(column);
					ctx.Column.InsertOnSubmit(entity);
					ctx.SubmitChanges();
				}
			}
			this.TestContext.WriteLine("Insert {0:#,0} rows done in {1:#,0}ms",
				cKeys, sw.ElapsedMilliseconds);

			// Verify that the data was written

			sw = Stopwatch.StartNew();
			for (int iKey = 0; iKey < cKeys; iKey++)
			{
				using (var ctx = CreateContext())
				{
					var key = string.Concat(keyPrefix, iKey);
					var rgentcol = ctx.Column
						.Where(x =>
							x.ColumnFamily == _ColumnFamily &&
							x.Key == key &&
							x.Column == columnName.ToCassandraByte());
					var value = string.Concat(valuePrefix, iKey);
					Assert.AreEqual(1, rgentcol.Count());
					foreach (var entcol in rgentcol)
					{
						Assert.AreEqual(key, entcol.Key.ToUTF8());
						Assert.AreEqual(columnName, entcol.Data.Name.ToUTF8());
						Assert.AreEqual(value, entcol.Data.Value.ToUTF8());
					}
				}
			}
			this.TestContext.WriteLine("Read {0:#,0} rows done in {1:#,0}ms",
				cKeys, sw.ElapsedMilliseconds);

			// Delete the data

			sw = Stopwatch.StartNew();
			using (var ctx = CreateContext())
			{
				for (int iKey = 0; iKey < cKeys; iKey++)
				{
					var key = string.Concat(keyPrefix, iKey);
					ctx.Column.DeleteOnSubmit(x =>
						x.ColumnFamily == _ColumnFamily &&
						x.Key == key &&
						x.Column == columnName.ToCassandraByte());
				}
				ctx.SubmitChanges();
			}
			this.TestContext.WriteLine("Delete {0:#,0} rows done in {1:#,0}ms",
				cKeys, sw.ElapsedMilliseconds);

			// Verify that the data was deleted (Column)

			sw = Stopwatch.StartNew();
			using (var ctx = CreateContext())
			{
				for (int iKey = 0; iKey < cKeys; iKey++)
				{
					var key = string.Concat(keyPrefix, iKey);
					var rgentcol = ctx.Column
						.Where(x =>
							x.ColumnFamily == _ColumnFamily &&
							x.Key == key &&
							x.Column == columnName.ToCassandraByte());
					if (rgentcol.Count() != 0)
					{
						var rgentcolWithoutValue = rgentcol
							.ToArray()
							.Where(x => x.Data != null);
						Assert.AreEqual(0, rgentcol.Count());
					}
				}
			}
			this.TestContext.WriteLine("Delete verify done in {0:#,0}ms",
				sw.ElapsedMilliseconds);
        }

        #endregion

        #region Add/Delete (Objects)

        public class Product
		{
			public int Id { get; set; }
			public string Name { get; set; }

			public override bool Equals(object obj)
			{
				var p = (Product)obj;
				return
					this.Id.Equals(p.Id) &&
					this.Name.Equals(p.Name);
			}

			public override int GetHashCode()
			{
				return this.Id.GetHashCode();
			}
		}

		[TestMethod]
		public void AddDeleteObjectTest()
		{
			// Insert a data

			var product = new Product()
			{
				Id = 123,
				Name = "TestProduct",
			};
			const string key = "CassandraContextTestKey";
			using (var ctx = CreateContext())
			{
				var entrgcol = new CassandraEntity<List<Column>>()
					.SetColumnFamily(_ColumnFamily)
					.SetKey(key)
					.SetData(product);
				ctx.ColumnList.InsertOnSubmit(entrgcol);
				ctx.SubmitChanges();
			}

			// Verify that the data was written

			Expression<Func<CassandraEntity<Column>, bool>> expcol = (x) =>
				x.ColumnFamily == _ColumnFamily &&
				x.Key == key;
			Expression<Func<CassandraEntity<List<Column>>, bool>> exprgcol = (x) =>
				x.ColumnFamily == _ColumnFamily &&
				x.Key == key;
			using (var ctx = CreateContext())
			{
				var rgentrgcol = ctx.ColumnList
					.Where(exprgcol);
				Assert.AreEqual(1, rgentrgcol.Count());
				var rgp = rgentrgcol
					.Select(entrgcolT => entrgcolT.ToObject<Product>());
				Assert.AreEqual(1, rgp.Count());
				foreach (var pT in rgp)
				{
					Assert.AreEqual(product, pT);
				}
			}

			// Delete the data

			using (var ctx = CreateContext())
			{
				ctx.Column.DeleteOnSubmit(expcol);
				ctx.SubmitChanges();
			}

			// Verify that the data was deleted (ColumnList)

			using (var ctx = CreateContext())
			{
				var rgentrgcol = ctx.ColumnList
					.Where(exprgcol);
				Assert.AreEqual(1, rgentrgcol.Count());
				Assert.AreEqual(0, rgentrgcol.First().Data.Count);
				var rgp = rgentrgcol
					.Select(entrgcolT => entrgcolT.ToObject<Product>());
				Assert.AreEqual(1, rgp.Count());
				Assert.IsNull(rgp.First());
			}
        }

        #endregion

        #region Deletes

        [TestMethod]
        // Deletion does not yet support SliceRange predicates.
        [ExpectedException(typeof(Apache.Cassandra.InvalidRequestException))]
        public void DeleteCountTest()
        {
            var key = "DeleteCountTest".ToCassandraByte();
            using (var ctx = CreateContext())
            {
                var rgcol = Enumerable.Range(0, 100)
                    .Select(i => ((long)i).ToCassandraByte())
                    .Select(columnName => new Column()
                        .SetNameValue(columnName, columnName));
                ctx.ColumnList.InsertOnSubmit(new CassandraEntity<List<Column>>()
                {
                    ColumnFamily = _ColumnFamily,
                    Key = key,
                    Data = rgcol.ToList(),
                });
                ctx.SubmitChanges();
            }

            using (var ctx = CreateContext())
            {
                var deletion = new Deletion
                {
                    Predicate = new SlicePredicate()
                    {
                        Slice_range = new SliceRange()
                        {
                            Start = new byte[0],
                            Finish = new byte[0],
                            Count = 30,
                            Reversed = false,
                        }
                    }
                };
                ctx.MutationMap.Add(_ColumnFamily, key, deletion);
                ctx.SubmitChanges();
            }
        }

        [TestMethod]
        // Deletion.Slice_Predicate is not supported in Cassandra 0.6
        // http://cassandra-user-incubator-apache-org.3065146.n2.nabble.com/batch-mutate-deletion-slice-range-predicate-unsupported-td5048309.html
        // http://mail-archives.apache.org/mod_mbox/cassandra-user/201101.mbox/%3CAANLkTimOyGZ3cn=jjhirZ9uZCukLKm37=b2TYuXXSFaT@mail.gmail.com%3E
        // https://issues.apache.org/jira/browse/CASSANDRA-494
        // Deletion does not yet support SliceRange predicates.
        [ExpectedException(typeof(Apache.Cassandra.InvalidRequestException))]
        public void DeleteBetweenTest()
        {
            var key = "Key1".ToCassandraByte();

            using (var ctx = CreateContext())
            {
                ctx.Column.DeleteOnSubmit(x =>
                    x.ColumnFamily == _ColumnFamily &&
                    x.Key == key);
                ctx.SubmitChanges();

                var range = Enumerable.Range(1, 1000).ToArray();
                foreach (var i in range)
                    ctx.InsertOnSubmit(_ColumnFamily, key, new Column().SetNameValue(i, i));
                ctx.SubmitChanges();

                var actual = ctx.ColumnList
                    .Where(x =>
                        x.ColumnFamily == _ColumnFamily &&
                        x.Key == key)
                    .TakeColumn(int.MaxValue)
                    .SelectMany(x => x.Data)
                    .Select(c => c.Name.ToInt32())
                    .ToArray();
                CollectionAssert.AreEqual(range, actual);

                ctx.ColumnList.DeleteOnSubmit(x =>
                    x.ColumnFamily == _ColumnFamily &&
                    x.Key == key &&
                    x.Column.Between(0, 500));
                ctx.SubmitChanges();
            }
        }

        [TestMethod]
        // System.InvalidOperationException: You can use 'In' method belong [SuperColumn, Column] only in query condition.
        [ExpectedException(typeof(InvalidOperationException))]
        public void DeleteKeyInTest()
        {
            using (var ctx = CreateContext())
            {
                var range = Enumerable.Range(1, 1000).ToArray();
                foreach (var i in range)
                    ctx.InsertOnSubmit(_ColumnFamily, i, new Column().SetNameValue(i, i));
                ctx.SubmitChanges();

                ctx.ColumnList.DeleteOnSubmit(e =>
                    e.ColumnFamily == _ColumnFamily &&
                    e.Key.In(range));
                ctx.SubmitChanges();
            }
        }

        #endregion

        #region Mutations

        [TestMethod]
        public void RemoveIfNoChangeTest()
        {
			var key = "RemoveIfNoChangeTest".ToCassandraByte();
            CassandraContextTest.DeleteKeyAndSubmitChanges(key);
            try
            {
                using (var ctx = CassandraContextTest.CreateContext())
                {
					// Insert (count) columns.

                    const int count = 100;
                    var rgcol = Enumerable.Range(0, count)
                        .Select(i => ((long)i).ToCassandraByte())
                        .Select(rgb => new Column()
                            .SetNameValue(rgb, rgb))
                        .ToList();
					ctx.InsertOnSubmit(_ColumnFamily, key, rgcol);
                    ctx.SubmitChanges();
                    Assert.AreEqual(0, ctx.MutationsCount);

					// Try to insert the same list.
					// All such mutations should be removed.

					ctx.InsertOnSubmit(_ColumnFamily, key, rgcol);
					Assert.AreEqual(1, ctx.MutationMap.Count);
                    Assert.AreEqual(count, ctx.MutationMap[key][_ColumnFamily].Count);
                    ctx.RemoveMutationsIfNoChanges();
                    Assert.AreEqual(0, ctx.MutationMap.Count);

					// Try to insert the same list again,
					// but this time columns are re-created
					// to make sure they have different Timestamp.
					// All such mutations should be removed too.

					rgcol = Enumerable.Range(0, count)
						.Select(i => ((long)i).ToCassandraByte())
						.Select(rgb => new Column()
							.SetNameValue(rgb, rgb))
						.ToList();
					ctx.InsertOnSubmit(_ColumnFamily, key, rgcol);
					Assert.AreEqual(1, ctx.MutationMap.Count);
					Assert.AreEqual(count, ctx.MutationMap[key][_ColumnFamily].Count);
					ctx.RemoveMutationsIfNoChanges();
					Assert.AreEqual(0, ctx.MutationMap.Count);

					// Change the value of one column.
					// That column should be written to the DB,
					// while the rests were removed.

                    rgcol[30].SetValue(((long)31).ToCassandraByte());
					ctx.InsertOnSubmit(_ColumnFamily, key, rgcol);
                    Assert.AreEqual(1, ctx.MutationMap.Count);
                    Assert.AreEqual(count, ctx.MutationMap[key][_ColumnFamily].Count);

                    ctx.RemoveMutationsIfNoChanges();
                    Assert.AreEqual(1, ctx.MutationMap.Count);
                    Assert.AreEqual(1, ctx.MutationMap[key][_ColumnFamily].Count);
                }
            }
            finally
            {
                CassandraContextTest.DeleteKeyAndSubmitChanges(key);
            }
        }

		[TestMethod]
		public void RemoveIfNoChangeSuperColumnTest()
		{
			var key = "RemoveIfNoChangeTest".ToCassandraByte();
			CassandraContextTest.DeleteKeyAndSubmitChanges(key);
			try
			{
				using (var ctx = CassandraContextTest.CreateContext())
				{
					// Insert a SuperColumn with (count) columns.

					const int count = 100;
					var rgcolSource = Enumerable.Range(0, count)
						.Select(i => ((long)i).ToCassandraByte())
						.Select(rgb => new Column()
							.SetNameValue(rgb, rgb));
					var rgcol = rgcolSource.ToList();
					var sco = new SuperColumn
					{
						Name = 0L.ToCassandraByte(),
						Columns = rgcol,
					};
					ctx.InsertOnSubmit(_SuperColumnFamily, key, sco);
					ctx.SubmitChanges();
					Assert.AreEqual(0, ctx.MutationsCount);

					// Try to insert it again.
					// All such mutations should be removed.

					rgcol = rgcolSource.ToList();
					sco = new SuperColumn
					{
						Name = 0L.ToCassandraByte(),
						Columns = rgcol,
					};
					ctx.InsertOnSubmit(_SuperColumnFamily, key, sco);
					Assert.AreEqual(1, ctx.MutationMap.Count);
					var rgmut = ctx.MutationMap[key][_SuperColumnFamily];
					Assert.AreEqual(1, rgmut.Count);
					ctx.RemoveMutationsIfNoChanges();
					Assert.AreEqual(0, ctx.MutationMap.Count);

					// Try to insert the same list again,
					// but this time columns are re-created
					// to make sure they have different Timestamp.
					// All such mutations should be removed too.

					rgcol = rgcolSource.ToList();
					sco = new SuperColumn
					{
						Name = 0L.ToCassandraByte(),
						Columns = rgcol,
					};
					ctx.InsertOnSubmit(_SuperColumnFamily, key, sco);
					ctx.RemoveMutationsIfNoChanges();
					Assert.AreEqual(0, ctx.MutationMap.Count);

					// Change the value of one column.
					// That column should be written to the DB,
					// while the rests were removed.

					rgcol = rgcolSource.ToList();
					sco = new SuperColumn
					{
						Name = 0L.ToCassandraByte(),
						Columns = rgcol,
					};
					rgcol[30].SetValue(((long)31).ToCassandraByte());
					ctx.InsertOnSubmit(_SuperColumnFamily, key, sco);
					Assert.AreEqual(1, ctx.MutationMap.Count);
					rgmut = ctx.MutationMap[key][_SuperColumnFamily];
					Assert.AreEqual(1, rgmut.Count);
					Assert.AreEqual(count, rgmut[0].Column_or_supercolumn.Super_column.Columns.Count);

					ctx.RemoveMutationsIfNoChanges();
					Assert.AreEqual(1, ctx.MutationMap.Count);
					rgmut = ctx.MutationMap[key][_SuperColumnFamily];
					Assert.AreEqual(1, rgmut.Count);
					Assert.AreEqual(1, rgmut[0].Column_or_supercolumn.Super_column.Columns.Count);
				}
			}
			finally
			{
				CassandraContextTest.DeleteKeyAndSubmitChanges(key);
			}
		}

        #endregion

        #region Timestamp

        [TestMethod]
        public void TimestampTest()
        {
            var key = "0".ToCassandraByte();
            const string columnName = "name";
            using (var ctx = CassandraContextTest.CreateContext())
            {
                // Delete and then insert.

                var column = new Column()
                    .SetNameValue(columnName, "value");
                ctx.Column.DeleteOnSubmit(x =>
                    x.ColumnFamily == _ColumnFamily &&
                    x.Key == key &&
                    x.Column == columnName.ToCassandraByte());
                Thread.Sleep(1); // current timestamp resolution requires this
                ctx.InsertOnSubmit(_ColumnFamily, key, column);

                var del = ctx.MutationMap.Mutations()
                    .Select(m => m.Deletion)
                    .First(d => d != null);
                Assert.IsTrue(column.Timestamp > del.Timestamp);

                ctx.SubmitChanges();

                // Read the data.

                var result = ctx.Column.Where(x =>
                    x.ColumnFamily == _ColumnFamily &&
                    x.Key == key &&
                    x.Column == columnName.ToCassandraByte())
                    .FirstOrDefault();
                Assert.IsNotNull(result);
                Assert.IsNotNull(result.Data);

                // Next, insert and then delete.

                column = new Column()
                    .SetNameValue(columnName, "value");
                ctx.InsertOnSubmit(_ColumnFamily, key, new Column()
                    .SetNameValue(columnName, "value"));
                Thread.Sleep(1); // current timestamp resolution requires this
                ctx.Column.DeleteOnSubmit(x =>
                    x.ColumnFamily == _ColumnFamily &&
                    x.Key == key &&
                    x.Column == columnName.ToCassandraByte());

                del = ctx.MutationMap.Mutations()
                    .Select(m => m.Deletion)
                    .First(d => d != null);
                Assert.IsTrue(del.Timestamp > column.Timestamp);

                ctx.SubmitChanges();

                // Read the data.

                result = ctx.Column.Where(x =>
                    x.ColumnFamily == _ColumnFamily &&
                    x.Key == key &&
                    x.Column == columnName.ToCassandraByte())
                    .FirstOrDefault();
                Assert.IsNull(result);
            }
        }

        #endregion
    }
}
